package com.feiwo.nonego.rpc.consumer.common.handler;

import com.feiwo.nonego.rpc.consumer.common.context.RpcContext;
import com.feiwo.nonego.rpc.proxy.api.future.RpcFuture;
import com.feiwo.nonego.rpc.protocol.RpcProtocol;
import com.feiwo.nonego.rpc.protocol.request.RpcRequest;
import com.feiwo.nonego.rpc.protocol.response.RpcResponse;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 *
 *
 * @author feiwo 2025-08-08
 */
public class RpcConsumerHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcResponse>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumerHandler.class);

    private volatile Channel channel;

    private SocketAddress socketAddress;

    private static final Map<Long, RpcFuture> rpcFuturePeddingMap = new ConcurrentHashMap<>();

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        this.socketAddress = this.channel.remoteAddress();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol<RpcResponse> msg) throws Exception {
        if (Objects.isNull(msg) || Objects.isNull(msg.getHeader())) {
            return;
        }
        var header = msg.getHeader();
        var rpcFuture = rpcFuturePeddingMap.remove(header.getMsgId());
        if (Objects.nonNull(rpcFuture)) {
            rpcFuture.done(msg);
        }
    }

    public Optional<RpcFuture> sendRequest(RpcProtocol<RpcRequest> requestRpcProtocol, boolean async, boolean oneway) {
        return oneway ? this.sendRequestOneway(requestRpcProtocol) :
                async ? this.sendRequestAsync(requestRpcProtocol) :
                        this.sendRequestSync(requestRpcProtocol);
    }

    public Optional<RpcFuture> sendRequestSync(RpcProtocol<RpcRequest> requestRpcProtocol) {
        var rpcFuture = this.getRpcFuture(requestRpcProtocol);
        channel.writeAndFlush(requestRpcProtocol);
        return Optional.of(rpcFuture);
    }

    public Optional<RpcFuture> sendRequestAsync(RpcProtocol<RpcRequest> requestRpcProtocol) {
        var rpcFuture = this.getRpcFuture(requestRpcProtocol);
        RpcContext.getContext().setFuture(rpcFuture);
        channel.writeAndFlush(requestRpcProtocol);
        return Optional.empty();
    }

    public Optional<RpcFuture> sendRequestOneway(RpcProtocol<RpcRequest> requestRpcProtocol) {
        channel.writeAndFlush(requestRpcProtocol);
        return Optional.empty();
    }

    public void close() {
        this.channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    private RpcFuture getRpcFuture(RpcProtocol<RpcRequest> requestRpcProtocol) {
        var rpcFuture = new RpcFuture(requestRpcProtocol);
        var header = requestRpcProtocol.getHeader();
        rpcFuturePeddingMap.put(header.getMsgId(), rpcFuture);
        return rpcFuture;
    }


    public Channel getChannel() {
        return channel;
    }

    public SocketAddress getSocketAddress() {
        return socketAddress;
    }
}
